#ifndef __M_MSG_H__
#define __M_MSG_H__
#include "../mqcommon/helper.hpp"
#include "../mqcommon/log.hpp"
#include "../mqcommon/mq_meg.pb.h"

#include <iostream>
#include <string>
#include <unordered_map>
#include <memory>
#include <vector>
#include <cassert>
#include <list>

namespace pprmq
{
    using MessagePtr = std::shared_ptr<Message>;
    #define DATAFILE_SUBFIX ".mqd"
    #define TMPFILE_SUBFIX ".mqd.tmp" 
    // 持久化存储，消息的持久化存储不是存储在数据库中的，而是文件中的
    class MessageMapper
    {
    public:
        MessageMapper(std::string basedir,std::string qname):_qname(qname)
        {
            // 构建文件路径
            if(basedir.back() != '/') basedir.push_back('/');
            _datafile = basedir + qname + DATAFILE_SUBFIX;
            _tmpfile = basedir + qname + TMPFILE_SUBFIX;
            // 确保基础目录存在
            if(pprmq::FileHelper(basedir).exists() == false) 
                assert(pprmq::FileHelper::createDirectory(basedir));
            // 初始化数据文件
            createMsgFile();
        }

        // 创建消息持久化文件
        bool createMsgFile()
        {
            // 若数据文件已存在，直接返回成功
            if(pprmq::FileHelper(_datafile).exists())
            {
                return true;
            }
            // 尝试创建数据文件
            bool ret = FileHelper::createFile(_datafile);
            if(ret == false)
            {
                DLOG("创建数据文件%s失败",_datafile.c_str());
                return false;
            }
            return true;
        }

        // 删除消息相关文件（数据文件和临时文件）
        void removeMsgFile()
        {
            pprmq::FileHelper::removeFile(_datafile);
            pprmq::FileHelper::removeFile(_tmpfile);    
        }

        // 向数据文件插入消息（调用私有insert重载方法）
        bool insert(MessagePtr &msg)
        {
            return insert(_datafile,msg);
        }

        // 标记消息为无效,也就是逻辑上删除,并没有真正删除
        bool remove(MessagePtr &msg)
        {
            // 1. 将消息的有效标志位设置为无效
            msg->mutable_payload()->set_valid("0");
            // 2. 序列化更新后的消息体
            std::string body = msg->payload().SerializeAsString();
            // 校验序列化后的数据长度是否与原长度一致（避免破坏文件结构）
            if(msg->len() != body.size())
            {
                DLOG("不能修改文件中的数据信息，因为新生成的数据与原数据长度不一致");
                return false;
            }
            // 3. 覆盖写入到原消息在文件中的位置
            pprmq::FileHelper helper(_datafile);
            bool ret = helper.write(body.c_str(),msg->offset(),body.size());
            if(ret == false)
            {
                DLOG("向队列数据文件写入数据失败");
                return false;
            }
            return true;
        }
        
        // 垃圾回收机制
        std::list<MessagePtr> gc()
        {
            // 有效数据
            std::list<MessagePtr> result;
            // 1. 加载所有有效数据到result
            bool ret = load(result);
            if(ret == false)
            {
                DLOG("加载有效数据失败");
                return result;
            }
            // 2. 创建临时文件，将有效数据写入临时文件中
            FileHelper::createFile(_tmpfile);
            for(auto& msg : result)
            {
                ret = insert(_tmpfile,msg);
                if(ret == false)
                {
                    DLOG("向临时文件写入消息数据失败");
                    return result;
                }
            }
            // 3. 删除原数据文件
            ret = FileHelper::removeFile(_datafile);
            if(ret == false)
            {
                DLOG("删除源文件失败");
                return result;
            }
            // 4. 修改临时文件名
            ret = FileHelper(_tmpfile).rename(_datafile);
            if (ret == false) 
            {
                DLOG("修改临时文件名称失败！");
                return result;
            }
            //5. 返回新的有效数据
            return result;
        }

    private:
        // 从数据文件加载所有有效消息到列表
        bool load(std::list<MessagePtr> &result)
        {
            FileHelper data_file_helper(_datafile);
            // 读取位置，消息长度，文件总大小
            size_t offset = 0,msg_size = 0,file_size = data_file_helper.size();

            // while读取，只要还没读到文件末尾，就一直读
            while(offset < file_size)
            {
                // 读取逻辑
                // 1. 读消息长度
                bool ret = data_file_helper.read((char*)&msg_size,offset,sizeof(size_t));
                if(ret == false)
                {
                    DLOG("读取消息长度失败");
                    return false;
                }
                offset += sizeof(size_t);
                // 2. 读消息内容
                    // 准备一个能装下msg_size字节的空容器
                    // 从当前位置(offset)读msg_size字节，存到msg_body里
                std::string msg_body(msg_size,'\0');
                ret = data_file_helper.read(&msg_body[0],offset,msg_size);
                if(ret == false)
                {
                    DLOG("读取消息内容失败");
                    return false;
                }
                offset += msg_size;
                // 3. 解析消息内容
                    // 创建一个空消息对象
                    // 把刚才读到的二进制内容（msg_body）转成消息对象，也就是反序列化
                MessagePtr msgp = std::make_shared<Message>();
                msgp->mutable_payload()->ParseFromString(msg_body);
                // 4.过滤无效消息
                if(msgp->payload().valid() == "0")
                {
                    DLOG("加载到无效消息:%s", msgp->payload().body().c_str());
                    continue;
                }
                // 5. 将有效消息放入结果
                result.push_back(msgp);
            }
            return true;
        }

        // 向指定文件插入消息
        bool insert(const std::string &filename, MessagePtr &msg)
        {
            // 1. 序列化消息体
            std::string body = msg->payload().SerializeAsString();
            pprmq::FileHelper helper(filename);
            // 2. 获取文件当前大小（作为写入起始位置）
            size_t fsize =  helper.size();
            size_t msg_size = body.size();
            // 写入逻辑：先写消息长度，再写消息体       [消息长度(4字节)][消息体]
            // 必须通过&msg_size获取地址，才能让write方法读取msg_size的完整二进制数据!
            int ret = helper.write((char*)&msg_size,fsize,sizeof(size_t));
            if(ret == false)
            {
                DLOG("向队列数据文件写入数据长度失败");
                return false;
            }

            ret = helper.write(body.c_str(),fsize + sizeof(size_t),body.size());
            if(ret == false)
            {
                DLOG("向队列数据文件写入数据失败");
                return false;
            }
            // 3. 更新消息的存储信息（在文件中的偏移量和长度）
            msg->set_offset(fsize + sizeof(size_t));
            msg->set_len(body.size());
            return true;
        }
    private:
        std::string _qname;
        std::string _datafile;
        std::string _tmpfile;
    };

    class QueueMessage
    {
    public:
        using ptr = std::shared_ptr<QueueMessage>;
        QueueMessage(std::string &basedir, const std::string &qname)
            :_mapper(basedir,qname),_qname(qname),_valid_count(0),_total_count(0)
        {}

        // 恢复历史消息,从持久化存储中恢复未被确认删除的消息到内存
        bool recovery()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 调用_mapper的gc()获取存储中所有有效消息
            _msgs = _mapper.gc();
            // 遍历恢复的消息，将持久化消息存入_durable_msgs哈希表,按消息ID索引
            for(auto& msg : _msgs)
            {
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(),msg));
            }
            // 更新持久化消息计数：有效消息数和总消息数均等于恢复的消息量
            _valid_count = _total_count = _msgs.size();
            return true;
        }

        // 参数：bp（消息属性，含ID、持久化模式等） body（消息体）   queue_is_durable（队列是否持久化）
        bool insert(const BasicProperties *bp, const std::string &body,bool queue_is_durable)
        {
            // 1. 构造消息对象
            MessagePtr msg = std::make_shared<Message>();
            // 设置消息体内容
            msg->mutable_payload()->set_body(body);
            // 根据是否有外部属性（bp）初始化消息属性
            if(bp != nullptr)
            {
                DeliveryMode mode = queue_is_durable ? bp->delivery() : DeliveryMode::UNDURABLE;
                // 有外部属性，将外部属性设置进去
                msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                msg->mutable_payload()->mutable_properties()->set_delivery(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            }
            else 
            {
                DeliveryMode mode = queue_is_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                // 没有外部属性，自动生成属性
                msg->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
                msg->mutable_payload()->mutable_properties()->set_delivery(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key("");
            }
            // 加锁保证线程安全
            std::unique_lock<std::mutex> lock(_mutex);

            // 2. 判断消息是否需要持久化
            if(msg->payload().properties().delivery() == DeliveryMode::DURABLE)
            {
                // 标记消息在持久化存储中为有效
                msg->mutable_payload()->set_valid("1");
                // 3. 调用_mapper持久化存储消息
                bool ret = _mapper.insert(msg);
                if(ret == false)
                {
                    DLOG("持久化存储消息：%s 失败了", body.c_str());
                    return false;
                }
                _total_count++;
                _valid_count++;
                // 将持久化消息存入哈希表
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(),msg));
            }
            // 4. 无论是否持久化，均加入内存待推送列表（供消费者获取）
            _msgs.push_back(msg);
            return true;
        }

        // 获取队首消息，从待推送列表取出消息，移至待确认列表，等待消费者确认删除
        MessagePtr front()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 若待推送列表为空，返回空指针
            if(_msgs.size() == 0)
            {
                return MessagePtr();
            }
            // 取出队首消息
            MessagePtr result = _msgs.front();
            // 从待推送列表中移除（已被消费者获取，不再重复推送）
            _msgs.pop_front();
            // 将消息存入待确认列表（key为消息ID），等待消费者调用 remove() 确认删除
            _waitack_msgs.insert(std::make_pair(result->payload().properties().id(),result));
            return result;
        }

        // 确认删除消息：从待确认列表移除，若为持久化消息则同时删除存储中的数据，触发GC检查
        // 根据消息的唯一ID进行删除
        bool remove(const std::string &msg_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 1. 从待确认列表中查找消息
            auto it = _waitack_msgs.find(msg_id);
            if(it == _waitack_msgs.end())
            {
                // 没找到，有可能是重复操作
                DLOG("没有找到要删除的消息 %s", msg_id.c_str());
                return true;
            }
            // 2. 若为持久化消息，需同步删除存储中的数据
            if(it->second->mutable_payload()->mutable_properties()->delivery() == DeliveryMode::DURABLE)
            {
                _mapper.remove(it->second);
                _durable_msgs.erase(msg_id);
                _valid_count--;
                gc();
            }
            // 3. 从待确认列表中移除
            _waitack_msgs.erase(msg_id);
            return true;
        }

        // 获取待推送消息数量
        size_t getable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _msgs.size();
        }

        // 获取总持久化消息数量
        size_t total_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_count;
        }
        // 获取当前内存中管理的持久化消息数量
        size_t durable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_msgs.size();
        }

        // 获取待确认消息数量
        size_t waitack_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitack_msgs.size();
        }
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.removeMsgFile();

            _msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();

            _valid_count = 0;
            _total_count = 0;
        }
    private:
        bool GCCheck()
        {
            // 总消息大于2000条，有效消息比例小于50%
            if(_total_count > 2000 && _valid_count * 10  / _total_count < 5)
            {
                return true;
            }
            return false;
        }

        void gc()
        {
            if(GCCheck() == false) return;
            // 调用_mapper的gc()清理存储中的无效消息，返回剩余的有效消息列表
            std::list<MessagePtr> msgs = _mapper.gc();
            // 遍历有效消息，同步更新内存中的消息状态
            for(auto& msg : msgs)
            {
                // 查找该消息是否在内存的持久化消息哈希表中
                auto it = _durable_msgs.find(msg->payload().properties().id());
                if(it == _durable_msgs.end())
                {
                    // 若内存中不存在，说明之前可能遗漏管理，重新加入待推送列表和持久化哈希表
                    DLOG("垃圾回收后，有一条持久化消息，在内存中没有进行管理");
                    _msgs.push_back(msg);
                    _durable_msgs.insert(std::make_pair(msg->payload().properties().id(),msg));
                    continue;
                }
                // 若内存中存在，更新消息在存储中的偏移量和长度,GC后存储位置可能变化
                it->second->set_offset(msg->offset());
                it->second->set_len(msg->len());
            }
            // 更新持久化消息计数：有效消息数和总消息数均等于GC后剩余的有效消息量
            _valid_count = _total_count = msgs.size();
        }

    private:
        std::mutex _mutex;                 
        // 队列名称：标识当前管理的队列
        std::string _qname;  

        size_t _valid_count;               
        size_t _total_count;               
        // 消息持久化工具：负责消息的持久化存储和垃圾回收
        MessageMapper _mapper;              
        // 待推送消息列表：消费者尚未获取的消息
        std::list<MessagePtr> _msgs;       
        // 持久化消息哈希表：按消息ID索引所有内存中管理的持久化消息
        std::unordered_map<std::string, MessagePtr> _durable_msgs; 
        // 待确认消息哈希表：按消息ID索引已被消费者获取但未确认删除的消息
        std::unordered_map<std::string, MessagePtr> _waitack_msgs;  
    };

    class MessageManager
    {   
    public:
        using ptr = std::shared_ptr<MessageManager>;
        // 构造函数：初始化消息存储的基础目录
        MessageManager(const std::string &basedir): _basedir(basedir){}

        // 清空所有队列中的消息
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for(auto& queue : _queue_msgs)
            {
                queue.second->clear();
            }
        }

        // 初始化指定名称的消息队列,不存在就创建
        void initQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it != _queue_msgs.end())
                {
                    // 已经存在
                    return;
                }
                // 不存在创建新队列
                qmp = std::make_shared<QueueMessage>(_basedir,qname);
                // 将新队列插入到管理的队列集合中
                _queue_msgs.insert(std::make_pair(qname,qmp));
            }
            // 解锁后执行队列的恢复操作，可能从磁盘加载历史消息，recovery本来是在QueueMessage中就进行恢复的
            // 但是recovery中也有其自己的锁，那样就会发生锁冲突，所以将recovery独立开来，在不同的作用域中
            // 直接调用
            qmp->recovery();
        }
        // 销毁指定名称的消息队列
        void destroyQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    // 没不找到，也就是不存在直接返回
                    return;
                }
                qmp = it->second;
                // 找到了从管理集合中移除
                _queue_msgs.erase(qname);
            }
            qmp->clear();
        }
        // 向指定队列插入一条消息
        bool insert(const std::string &qname, BasicProperties *bp, const std::string &body, bool queue_is_durable)
        {
            // if (bp == nullptr) 
            // {
            //     DLOG("BasicProperties 指针为空，插入消息失败");
            //     return false;
            // }
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("向队列%s新增消息失败,没有找到消息管理句柄", qname.c_str());
                    return false;
                }
                // 找到了
                qmp = it->second;
            }
            return qmp->insert(bp,body,queue_is_durable);
        }
        // 获取指定队列的队首消息（最早插入的消息）
        MessagePtr front(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("获取队列%s队首消息失败,没有找到消息管理句柄!", qname.c_str());
                    return MessagePtr();
                }
                qmp = it->second;
            }
            // 调用队列的front方法获取队首消息
            return qmp->front();
        }
        // 确认指定队列中某条消息已处理
        void ack(const std::string &qname, const std::string &msg_id)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if(it == _queue_msgs.end())
                {
                    DLOG("确认队列%s消息%s失败,没有找到消息管理句柄!", qname.c_str(), msg_id.c_str());
                    return ;
                }
                qmp = it->second;
            }
            // 调用队列的remove方法删除指定ID的消息,因为确认消息后就要在队列中删除该消息，避免重复传输
            qmp->remove(msg_id);
            return ;
        }
        // 获取指定队列中可获取的消息数量（待处理消息数）
        size_t getable_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) {
                    DLOG("获取队列%s待推送消息数量失败,没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->getable_count();
        }
        // 获取指定队列中总持久化消息数量（包括已处理和未处理）
        size_t total_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) {
                    DLOG("获取队列%s总持久化消息数量失败,没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->total_count();
        }
        // 获取指定队列中有效持久化消息数量（排除已删除的）
        size_t durable_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) {
                    DLOG("获取队列%s有效持久化消息数量失败,没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->durable_count();
        }
        // 获取指定队列中待确认的消息数量（已取出但未ack的消息）
        size_t waitack_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) {
                    DLOG("获取队列%s待确认消息数量失败,没有找到消息管理句柄!", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->waitack_count();
        }

    private:
        std::mutex _mutex;
        std::string _basedir; // 消息持久化的基础目录
        // 存储所有管理的消息队列：key为队列名称，value为队列对象的智能指针
        std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs;
    };
}
#endif